package com.lengxf.mqtt.config.inbound;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.lengxf.mqtt.channel.ChannelName;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
@AllArgsConstructor
public class MqttInboundFlow {

    private final MqttPahoMessageDrivenChannelAdapter mqttInboundAdapter;

    private final ObjectMapper objectMapper;

    /**
     * IntegrationFlows # from 绑定某个通道
     * IntegrationFlows # handle 请求处理
     * IntegrationFlows # transform 数据转换
     * IntegrationFlows # route 路由分发
     * IntegrationFlows # split 分流
     * IntegrationFlows # channel 转发
     *
     * @Author Lengxf
     */
    @Bean
    public IntegrationFlow mqttInbound() {
        return IntegrationFlow.from(mqttInboundAdapter)
                .<byte[]>handle((payload, headers) -> {
                    try {
                        String s = objectMapper.readValue(payload, String.class);
                        log.info("接收到消息 : {}", s);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
//                        return objectMapper.writeValueAsBytes("处理完成");
                    //若出参为null则会终结操作,不会发送至下游channel
                    return null;
                })
                .channel(ChannelName.OUTBOUND)
                .get();
    }

    private String handle(Message<?> m) {
        try {
            byte[] payload = (byte[]) m.getPayload();
            String s = objectMapper.readValue(payload, String.class);
            log.info("接收到消息 : {}", s);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return "处理完成...";
    }

}
